package cn.itcast.streaming.function.asyncio;

import cn.itcast.streaming.bean.ItcastDataPartObj;
import cn.itcast.streaming.bean.VehicelLocationInfo;
import cn.itcast.streaming.utils.GaoDeMapUtils;
import cn.itcast.streaming.utils.GeoHashUtil;
import cn.itcast.streaming.utils.RedisUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Supplier;

//{"status":"1","regeocode":{"addressComponent":{"city":[],"province":"北京市","adcode":"110101","district":"东城区","towncode":"110101001000","streetNumber":{"number":"4号","location":"116.395536,39.911223","direction":"西北","distance":"324.625","street":"西长安街"},"country":"中国","township":"东华门街道","businessAreas":[{"location":"116.416804,39.913479","name":"东单","id":"110101"},{"location":"116.412031,39.913505","name":"王府井","id":"110101"},{"location":"116.391134,39.895089","name":"大栅栏","id":"110102"}],"building":{"name":"天安门","type":"风景名胜;风景名胜相关;旅游景点"},"neighborhood":{"name":[],"type":[]},"citycode":"010"},"formatted_address":"北京市东城区东华门街道天安门"},"info":"OK","infocode":"10000"}


//异步请求高德地图api获取地理位置信息
public class AsyncHttpQueryFunction extends RichAsyncFunction<ItcastDataPartObj,ItcastDataPartObj> {

    private static final Logger logger = LoggerFactory.getLogger(AsyncHttpQueryFunction.class);
    //定义Httpclient对象
    CloseableHttpAsyncClient httpAsyncClient = null;

    //初始化资源
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //定义异步的httpclient的连接配置对象
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(3000)
                .setConnectTimeout(3000)
                .build();
        httpAsyncClient = HttpAsyncClients.custom()
                .setMaxConnTotal(10)
                .setDefaultRequestConfig(requestConfig) //设置异步的客户端连接器配置对象
                .build();
        //启动客户端连接对象
        httpAsyncClient.start();

    }

    //释放资源
    @Override
    public void close() throws Exception {
        super.close();
        httpAsyncClient.close();
    }

    //发送异步请求
    @Override
    public void asyncInvoke(ItcastDataPartObj itcastDataPartObj, ResultFuture<ItcastDataPartObj> resultFuture) throws Exception {

        //获取原始数据的金纬度
        double longtiude = itcastDataPartObj.getLng(); //经度
        double latitude = itcastDataPartObj.getLat();

        //创建高德api的请求地址
        String urlByLonLat = GaoDeMapUtils.getUrlByLonLat(longtiude, latitude);
        //todo 3.发送请求
        HttpPost httpPost = new HttpPost(urlByLonLat);
        //todo 4.递交请求，获取异步请求对象
        Future<HttpResponse> future = httpAsyncClient.execute(httpPost, null);//callback是回调函数，

        //从成功的future获取到数据，返回javabean对象
        CompletableFuture<ItcastDataPartObj> onlineDataObjCompletableFutrue = CompletableFuture
                .supplyAsync(new Supplier<ItcastDataPartObj>() {
            @Override
            public ItcastDataPartObj get() {
                try {
                    //使用future获取返回值
                    HttpResponse httpResponse = future.get();
                    String province = null;
                    String city = null;
                    String district = null;
                    String county = null;
                    String address = null;
                    if (httpResponse.getStatusLine().getStatusCode() == 200) {
                        //拿到相应的实例对象
                        HttpEntity entity = httpResponse.getEntity();
                        //将实例对象转换成string类型
                        String result = EntityUtils.toString(entity);
                        //将返回的字符串json转换为javabean
                        JSONObject jsonObject = JSON.parseObject(result);

                        //获取到regeocode节点的数据
                        JSONObject regeocodeObject = jsonObject.getJSONObject("regeocode");
                        if (regeocodeObject != null && !regeocodeObject.isEmpty()) {
                            JSONObject addressComponent = regeocodeObject.getJSONObject("addressComponent");
                            county = addressComponent.getString("country");
                            province = addressComponent.getString("province");
                            city = addressComponent.getString("city");
                            district = addressComponent.getString("district");
                            address = addressComponent.getString("formatted_address");

                            VehicelLocationInfo vehicelLocationInfo = new VehicelLocationInfo(
                                    county, province, city, district, address);

                            String geoHash = GeoHashUtil.encode(itcastDataPartObj.getLat(), itcastDataPartObj.getLng());
                            //将数据模型对象转换成字符串写入到redis中，以供下次查询
                            RedisUtil.set(geoHash.getBytes(), JSON.toJSONString(vehicelLocationInfo).getBytes());

                        }

                    }

                    //拉宽原始数据的javabean对象
                    itcastDataPartObj.setCountry(county);
                    itcastDataPartObj.setProvince(province);
                    itcastDataPartObj.setCity(city);
                    itcastDataPartObj.setDistrict(district);
                    itcastDataPartObj.setAddress(address);
                } catch (Exception e) {
                    System.out.println("获取高德地图api逆地理位置查询失败。。。" + e.getMessage());
                    e.printStackTrace();
                }
                return itcastDataPartObj;
            }
        });

        //todo 6:将获取到的数据存储到resultFuture对象中，返回给方法
        onlineDataObjCompletableFutrue.thenAccept(new Consumer<ItcastDataPartObj>() {
            @Override
            public void accept(ItcastDataPartObj itcastDataPartObj) {
                //complete里面需要的是collection集合，但是一次执行只能返回一个结果
                //所以集合使用singleton方法的单例模式，只存储一条数据进行返回
                resultFuture.complete(Collections.singleton(itcastDataPartObj));
            }
        });

    }


    //请求超时调用的方法，如果不重写请求超时会抛出异常
    @Override
    public void timeout(ItcastDataPartObj input, ResultFuture<ItcastDataPartObj> resultFuture) throws Exception {

        logger.error("访问高德地图逆地理位置查询的时候请求超时");

    }
}
